package com.gxljc.bear.crawler.itaogao.oeeee;

import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.bear.crawler.base.BaseMongoTools;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.itaogao.ItaogaoConst;
import com.gxljc.bear.crawler.itaogao.NewsPageTable;
import com.gxljc.bear.crawler.proxy.ProxyUtil;
import com.gxljc.bear.crawler.util.DateUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.io.Serializable;
import java.util.*;

/**
 * 南方都市报 seed 数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-9-22
 */
public class OeeeeUpdateCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(OeeeeUpdateCrawler.class);
    private Boolean proxy = false;
    OeeeeSeedTagCrawler crawler = null;
    OeeeeSeedCrawler seed = null;

    public OeeeeUpdateCrawler(Boolean proxy) {
        crawler = new OeeeeSeedTagCrawler(proxy);
        seed = new OeeeeSeedCrawler(proxy);
        this.proxy = proxy;
    }

    public void crawl() throws Exception {
        List<NewsPageTable> seeds = getSeed();
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        ProxyUtil.initQueue(proxy);
        if (seeds.size() < DolphinCrawlerConsts.SINGLE_CORE_TO_RUN_SIZE) {
            seed.crawlNoSpark(seeds);
        } else {
            seed.crawl(seeds);
        }
    }

    //构造URL seed
    public List<NewsPageTable> getSeed() throws Exception {
        Calendar now = Calendar.getInstance();
        Date date = now.getTime();
        String dateFormat = DateUtil.dateFormat(date, "yyyy-MM/dd");
        String url = String.format(ItaogaoConst.NEWSPAPER_SEED.OEEEE.getValue(), dateFormat);
        crawler.crawl(url);
        return getUrls();
    }

    //获取seed
    public List<NewsPageTable> getUrls() throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.ITAOGAO.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(DolphinCrawlerConsts.MONGODB_DEFAULT_ID, 1);
        keys.put(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue(), 1);
        DBObject query = new BasicDBObject();
        query.put(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_FLAG.getValue(), new BasicDBObject("$eq", 0));
        DBCursor cursor = repo.getCollection(ItaogaoConst.MONGODB_TABLE.OEEEE_SEED.getValue())
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<NewsPageTable> seeds = new LinkedList<>();
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object id = data.get(DolphinCrawlerConsts.MONGODB_DEFAULT_ID);
                Object url = data.get(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue());
                NewsPageTable table = new NewsPageTable();
                table.setId(id.toString());
                table.setCrawlUrl(url.toString());
                seeds.add(table);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return seeds;
    }
}
